package cn.getech.data.development.controller;


import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.config.properties.OozieConfig;
import cn.getech.data.development.constant.*;
import cn.getech.data.development.dto.JobLinkAndJobNodeDto;
import cn.getech.data.development.dto.JobLinkDto;
import cn.getech.data.development.dto.JobNodeInfoDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.model.dto.WorkMenuDto;
import cn.getech.data.development.model.vo.JobLinkVO;
import cn.getech.data.development.model.vo.JobNodeInfoVO;
import cn.getech.data.development.model.vo.SumOverViewVO;
import cn.getech.data.development.service.*;
import cn.getech.data.development.utils.HdfsUtil;
import cn.getech.data.development.utils.oozie.JobUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.PageUtils;
import cn.getech.data.intelligence.common.utils.PojoUtils;
import cn.getech.data.intelligence.common.utils.R;
import cn.getech.data.intelligence.common.utils.RegexUtils;
import cn.getech.system.center.annotation.SysLog;
import cn.getech.system.center.constant.OperationModulesEnum;
import cn.getech.system.center.constant.OperationTypeEnum;
import cn.getech.system.center.constant.module.ModuleEnum;
import cn.getech.system.center.entity.SysUserEntity;
import cn.getech.system.center.entity.SysUserTokenEntity;
import cn.getech.system.center.service.SysUserService;
import cn.getech.system.center.service.SysUserTokenService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.oozie.client.OozieClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.*;


/**
 * 任务信息
 *
 * @author zenith
 * @Date 2018-08-23 18:36:33
 */
@Slf4j
@Api(tags = "任务信息")
@RestController
@RequestMapping("/bdp/jobInfo")
public class JobInfoController {
    private static final Logger logger = LoggerFactory.getLogger(JobInfoController.class);


    @Autowired
    private JobInfoService jobInfoService;
    @Autowired
    private JobNodeInfoService jobNodeInfoService;
    @Autowired
    private ProcInfoService procInfoService;
    @Autowired
    private JobRunHistoryService jobRunHistoryService;
    @Autowired
    private SysUserService sysUserService;
    @Autowired
    private JobNodeConfService jobNodeConfService;
    @Autowired
    private BdpJobConfig bdpJobConfig;
    @Autowired
    private OozieConfig oozieConfig;
    @Autowired
    private JobNodeRunHistoryService jobNodeRunHistoryService;
    @Autowired
    private SysUserTokenService sysUserTokenService;

    @Autowired
    private IWorkMenuService workMenuService;
    @Autowired
    private IWorkMenuJobRelService workMenuJobRelService;

//    @Autowired
//    private ClouderaManagerCluster clouderaManagerCluster;

    @ApiOperation("概览")
    @GetMapping("/overview")
    @SysLog(value="概览",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.SYSMODECODE,platformModuleType = ModuleEnum.OPERATION_CENTER_MODULE_CODE)
    public R getOverview(){
        SumOverViewVO overView = jobInfoService.selectSumOverView();
//        clouderaManagerCluster.getAllService();
        return R.okWithData(overView);
    }


    /**
     * 列表
     * @return
     */
    @ApiOperation("获取该项目中的开发任务所有拥有者列表")
    @GetMapping("/jobOwnerList/{procId}")
    //@SysLog(value="获取该项目中的开发任务所有拥有者列表",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R jobOwnerList(@PathVariable("procId") Integer procId){
        return R.okWithData(this.jobInfoService.getJobOwnersList(procId));
    }

    /**
     * 列表
     *
     * @param params
     * @return
     */
    @ApiOperation("列表")
    @GetMapping("/list")
    //@SysLog(value="列表",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
//    @RequiresPermissions("bdp:jobInfo:list")
    public R list(@ApiParam(value = "{'name':'测试','proc_id':1,'state':1}") @RequestParam Map<String, Object> params) {
        params.put("sidx", "id");
        params.put("order", "desc");
        PageUtils page = jobInfoService.queryPage(params);
        for (JobInfo info : (List<JobInfo>) page.getList()
        ) {
            fillInfo(info);
        }
        return R.okWithPage(page);
    }

    private void fillInfo(JobInfo info) {
        //添加创建用户的信息
        JobNodeInfo jobNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("job_id", info.getId()).eq("type_id", 0));
        if (null != jobNodeInfo) {
            JobNodeConf expression = jobNodeConfService.selectKVByJobNodeIdAndKey(jobNodeInfo.getId(), "expression", 0);
            if (null != expression) {
                info.setStartNodeInfo(fillExpression(expression.getValue()));
            }
        }
        info.setCreatePerName(sysUserService.getById(info.getCreatePer()).getUsername());
        info.setEnableName(JobStatus.ObjOf(info.getEnable()).getName());
        info.setLastRunStateName(LastRunState.ObjOf(info.getLastRunState()).getName());
        info.setSuccessNum(jobRunHistoryService.count(new QueryWrapper<JobRunHistory>().eq("job_info_id", info.getId()).eq("state", 1)));
        info.setFailedNum(jobRunHistoryService.count(new QueryWrapper<JobRunHistory>().eq("job_info_id", info.getId()).eq("state", 0)));
    }

    /*
    任务节点配置expression字段转成可读性文字
     */
    private String fillExpression(String expression) {
        String startNodeInfo = null;
        if (null != expression && !"".equals(expression) && !"null".equals(expression)) {
            JSONObject jsonObj = JSONObject.parseObject(expression);
            String type = jsonObj.get("type").toString();

            switch (type) {
                case "hour": {
                    startNodeInfo = String.format("每小时 %s分", numToDouble(jsonObj.getString("minute")));
                }
                break;
                case "day": {
                    startNodeInfo = String.format("每天 %s:%s", numToDouble(jsonObj.getString("hour")), numToDouble(jsonObj.getString("minute")));
                }
                break;
                case "week": {
                    startNodeInfo = String.format("每周 周%s %s:%s", SevenDaysTypeEnum.ObjOf(jsonObj.getInteger("week")).getMsg(),
                            numToDouble(jsonObj.getString("hour")), numToDouble(jsonObj.getString("minute")));

                }
                break;
                case "month": {
                    startNodeInfo = String.format("每月%s号 %s:%s", jsonObj.getString("day"),
                            numToDouble(jsonObj.getString("hour")), numToDouble(jsonObj.getString("minute")));
                }
                break;
                default:
                    break;
            }
        } else if ("null".equals(expression))
            startNodeInfo = "手动触发";
        return startNodeInfo;
    }


    /**
     * 个位数转01，02，03形式
     *
     * @param num
     * @return
     */
    private String numToDouble(String num) {
        String numDouble = null;
        if (num.length() < 2) {
            numDouble = "0" + num;
            return numDouble;
        } else
            return num;

    }

    /**
     * 保存任务信息新增任务时创建一个开始的任务
     */
    @ApiOperation("保存任务信息")
    @SysLog(value="保存任务信息",type= OperationTypeEnum.ADDCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/save")
//    @RequiresPermissions("bdp:jobInfo:save")
    public R save(@ApiParam(value = "{'name':'测试','procId':1,'desc':'描述'}") @RequestBody JobInfo jobInfo) {
        if (jobInfoService.checkExistByNameAndProcId(jobInfo.getName(), jobInfo.getProcId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_EXISTED.getMessage());
        } else {
            //验证email
            if (null == jobInfo || (StringUtils.isNotEmpty(jobInfo.getFailMailAddress()))) {
                cheackEmails(jobInfo.getFailMailAddress());
            }
            jobInfo.setCreateTime(DateUtil.date());
            jobInfo.setCreatePer(ShiroUtils.getUserId().intValue());
            //禁用状态
            jobInfo.setEnable(false);
            jobInfo.setLastRunState(LastRunState.NOTRUN.getCode());
            ProcInfo procInfo = procInfoService.getById(jobInfo.getProcId());
            if (procInfo == null) {
                throw new RRException(DataDevelopmentBizExceptionEnum.PROCINFO_NOT_EXIST.getMessage());
            }
            jobInfoService.save(jobInfo);
            return R.ok();
        }
    }

    private void cheackEmails(String failMailAddress) {
        if (StringUtils.isEmpty(failMailAddress)) {
            return;
        }
        //验证email已逗号隔开
        String[] split = failMailAddress.split(",");
        if (null != split && split.length > 0) {
            for (String s : split) {
                if (!RegexUtils.checkEmail(s)) {
                    throw new RRException("邮箱格式错误！");
                }
            }
        }
    }

    /**
     * 修改任务信息
     */
    @ApiOperation("修改任务信息")
    @SysLog(value="修改任务信息",type= OperationTypeEnum.UPDATECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/update")
//    @RequiresPermissions("bdp:jobInfo:update")
    public R update(@ApiParam(value = "{'id':1,'procId':1,'desc':'描述'}") @RequestBody JobInfo param) {
        checkIsMod(param);
        Integer jobInfoId = param.getId();
        JobInfo jobInfo = jobInfoService.getById(jobInfoId);
        if (jobInfo.getEnable()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_ENABLE.getMessage());
        }
        jobInfo.setModTime(DateUtil.date());
        jobInfo.setModPer(ShiroUtils.getUserId().intValue());

        if (param.getDesc() != null) {
            jobInfo.setDesc(param.getDesc());
        }
        if (param.getName() != null) {
            jobInfo.setName(param.getName());
        }
        //验证email已逗号隔开
        if (StringUtils.isNotEmpty(param.getFailMailAddress())) {
            cheackEmails(param.getFailMailAddress());
        }
        jobInfo.setFailMailAddress(param.getFailMailAddress());
        jobInfoService.updateById(jobInfo);
        return R.ok();
    }

    /**
     * 删除任务
     */
    @ApiOperation("删除任务")
    @SysLog(value="删除任务",type= OperationTypeEnum.DELETECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/delete")
    //    @RequiresPermissions("bdp:jobInfo:delete")
    public R delete(@ApiParam(value = "{'id':1}") @RequestBody JobInfo param) throws Exception {
        checkIsMod(param);
        Integer jobInfoId = param.getId();
        Long userId = ShiroUtils.getUserId();
        JobInfo jobInfo = jobInfoService.getById(jobInfoId);
        if (jobInfo.getEnable()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_CANT_DELETE.getMessage());
        }
        if (jobInfoService.checkIfRef(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_NODE_DEPEND.getMessage());
        } else {
            jobInfoService.deleteAllOfJob(jobInfoId, userId);
            return R.ok();
        }
    }

    /**
     * 启用任务
     * 如果是定时任务的话生成coordinator.xml至hdfs并运行
     *
     * @return
     */
    @SysLog(value="启用任务",type= OperationTypeEnum.UPDATECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @ApiOperation("启用任务")
    @PostMapping("/enableJobInfo")
    //    @RequiresPermissions("bdp:jobInfo:enableJobInfo")
    public R enableJobInfo(@ApiParam(value = "{'id':1}") @RequestBody JobInfo param) throws Exception {
        checkIsMod(param);

        //验证只有一个开始节点配置、开始节点是否有其它的连线、所有节点是否配置参数(必须有结束节点)。还得设置关联
        jobNodeInfoService.checkJobConfigData(param);

        JobInfo jobInfo = jobInfoService.getById(param.getId());
        JobNodeInfo startNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", param.getId())
                .eq("type_id", 0));
        if (startNodeInfo == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_NODE_START_NO_CONF.getMessage());
        }
//        startNodeInfo.setEnable(true);
//        jobInfo.setEnable(true);
//        //判断是否为定时循环任务
//        JobNodeConf jobNodeConf = jobNodeConfService.getOne(new QueryWrapper<JobNodeConf>().eq("`key`", "expression").eq("job_node_id", startNodeInfo.getId()).eq("job_type",0));
//        JobNodeConf jobNodeConfScheduleOrInterval = jobNodeConfService.getOne(new QueryWrapper<JobNodeConf>().eq("`key`", "scheduleOrInterval").eq("job_node_id", startNodeInfo.getId()).eq("job_type",0));
//        if (jobNodeConf.getValue() != null && !jobNodeConf.getValue().equals("null")) {
//            String expression = jobNodeConf.getValue();
//            String scheduleOrIntervalCode = jobNodeConfScheduleOrInterval.getValue();
//            OozieClient wc = new OozieClient(oozieConfig.getUrl());
//            HdfsUtil hdfsUtil = new HdfsUtil(bdpJobConfig);
//            JobUtil jobUtil = new JobUtil(jobInfoService, hdfsUtil, wc);
//            Long userId = ShiroUtils.getUserId();
//
////            jobNodeInfoService.initJobHistoryAndJobNodeHistory(param, "定时运行初始化数据");
//
//            //保存执行回调函数的shell
//            Integer procId = jobInfo.getProcId();
//            Integer jobId = jobInfo.getId();
//            String jobName = procId.toString() + "/" + jobId.toString();
//
//            Properties conf = jobUtil.createOrUpdateJob(userId, param.getId(), jobInfo, true);
//            String oozieJobId = jobUtil.runCoordinatorJob(userId, jobInfo.getId(), conf, expression, scheduleOrIntervalCode);
//            String beginShell = "#!/bin/sh\n+" +
//                    "source /etc/profile\n" +
//                    "param=${time_hour}\n" +
//                    "stat_date=`date +%Y-%m-%d`\n" +
//                    "if [ -z \"$param\" ]; then\n" +
//                    "param=`date '+%Y-%m-%d %H:%M:%S'`\n" +
//                    "fi\n" +
//                    "function run(){  \necho 'end run'" +
//                    "\n}\n" +
//                    "run && (curl -H \"Content-Type:application/json;charset=UTF-8\" -d \"{\"\'\"jobName\"\'\":\\\"" + jobName + "\\\",\"\'\"oozieJobId\"\'\":\\\"" + oozieJobId + "\\\",\"\'\"stat_date\"\'\":\\\"${param}\\\" ,\"\'\"isCorJob\"\'\":\\\"" + true + "\\\"}\" \"" + oozieConfig.getRest_url() + "/job_begin\" &) " +
//                    "|| (curl -H \"Content-Type:application/json;charset=UTF-8\" -d \"{\"\'\"jobName\"\'\":\\\"" + jobName + "\\\",\"\'\"oozieJobId\"\'\":\\\"" + oozieJobId + "\\\",\"\'\"stat_date\"\'\":\\\"${param}\\\",\"\'\"isCorJob\"\'\":\\\"" + true + "\\\"}\" \"" + oozieConfig.getRest_url() + "/job_begin\" & exit 1)";
////            String restShell = String.format(shell,"/job_end");
////            String beginShell =  String.format(shell,"/job_begin");
////            String restDst = "workflow-app/" + userId.toString() + "/" + jobId.toString() + "/rest.sh";
//            String startDst = "workflow-app/" + userId.toString() + "/" + jobId.toString() + "/StartTwo" + startNodeInfo.getNodeKey().substring(1) + ".sh";
////            hdfsUtil.writeFile(restShell.getBytes(), restDst);
//            hdfsUtil.writeFile(beginShell.getBytes(), startDst);
//            jobInfo.setLastCorOozieJobId(oozieJobId);
//            jobInfo.setLastRunNum(1);
////            jobInfo.setLastRunState(LastRunState.RUNNING.getCode());
//
//            //更新jobRunHistory
////            JobRunHistory jobRunHistory = jobRunHistoryService.getBaseMapper().selectList(new QueryWrapper<JobRunHistory>().eq("job_info_id", jobInfo.getId()).orderByDesc("num")).get(0);
////            jobRunHistory.setOozieJobId(oozieJobId);
////            jobRunHistoryService.updateById(jobRunHistory);
//
////            JobNodeRunHistory jobNodeRunHistory = jobNodeRunHistoryService.getBaseMapper().selectList(new QueryWrapper<JobNodeRunHistory>().eq("job_info_id", jobInfo.getId()).orderByDesc("num")).get(0);
////            jobNodeRunHistory.setOozieJobId(oozieJobId);
////            jobNodeRunHistoryService.updateById(jobNodeRunHistory);
//
//        }
//        if (jobInfoService.updateById(jobInfo)) {
//            jobNodeInfoService.updateById(startNodeInfo);
//            return R.ok();
//        }
        return R.ok();
    }


    /**
     * 停止正在运行的任务
     *
     * @return
     */
    @SysLog(value="停止正在运行的任务",type= OperationTypeEnum.UPDATECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @ApiOperation("停止正在运行的任务")
    @PostMapping("/stopJobInfo")
    public R stopJobInfo(@ApiParam(value = "{'id':1}") @RequestBody JobInfo param) throws Exception {
        checkStopIsMod(param);
        JobInfo jobInfo = jobInfoService.getById(param.getId());
        if (null == jobInfo)
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NOTEXISTED.getMessage());
        jobInfoService.stopJobInfo(param.getId(), jobInfo.getLastOozieJobId());
        return R.ok();
    }


    /**
     * 禁用任务
     *
     * @return
     */
    @SysLog(value="禁用任务",type= OperationTypeEnum.OPENCLOSECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @ApiOperation("禁用任务")
    @PostMapping("/disableJobInfo")
    //    @RequiresPermissions("bdp:jobInfo:disableJobInfo")
    public R disableJobInfo(@ApiParam(value = "{'id':1}") @RequestBody JobInfo param) {
        if (jobInfoService.checkIsRun(param.getId())) {
            param.setLastRunState(0);
        }
        checkIsMod(param);
        JobInfo jobInfo = jobInfoService.getById(param.getId());
        if (null == jobInfo) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NOTEXISTED.getMessage());
        }
        //如果是定时任务，则需要停止定时任务
        JobNodeInfo startNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", param.getId())
                .eq("type_id", 0));
        JobNodeConf jobNodeConf = jobNodeConfService.getOne(new QueryWrapper<JobNodeConf>().eq("`key`", "expression").eq("job_node_id", startNodeInfo.getId()).eq("job_type",0));
        if (jobNodeConf.getValue() != null && !jobNodeConf.getValue().equals("null")) {
            try {
                jobInfoService.stopJobInfo(jobInfo.getId(), jobInfo.getLastCorOozieJobId());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }


        //停止当前job下的所有的任务
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>().eq("job_id", param.getId()));
        if (null == jobNodeInfos || jobNodeInfos.isEmpty()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_NODE_START_NO_CONF.getMessage());
        }
        Set<String> sList = new HashSet<>();
        jobNodeInfos.forEach(jobNodeInfo -> {
            sList.add(jobNodeInfo.getNodeKey());
        });

        //禁用其下所有的jenkins节点数据
        jobNodeInfoService.disabledOrabledJenkins(sList, jobNodeInfos, param.getId(), false);

        jobInfo.setEnable(false);
        jobInfoService.updateById(jobInfo);


//        JobNodeInfo jobNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>()
//                .eq("job_id",param.getId())
//                .eq("type_id",0));
//        if(jobNodeInfo==null){
//            throw new RRException("请设置开始节点");
//        }
//        if (!jobNodeInfo.getEnable()) {
//            throw new RRException(BizExceptionEnum.JOBINFO_DISABLE);
//        }
//        jobNodeInfo.setEnable(false);
//        jobInfo.setEnable(false);
//        if(jobInfoService.updateById(jobInfo)) {
//            if (jobNodeInfoService.updateById(jobNodeInfo)) {
//                JobNodeUtil jobNodeUtil = new JobNodeUtil(String.valueOf(jobInfo.getProcId()), String.valueOf(jobInfo.getId()), jenkinsConfig);
//                try {
//                    jobNodeUtil.disableJob(jobNodeInfo.getNodeKey());
//                    return R.ok();
//                } catch (Exception e) {
//                    throw new RRException(BizExceptionEnum.SERVER_ERROR);
//                }
//            }
//        }
        return R.ok();
    }

    /**
     * 执行任务
     */
    @SysLog(value="执行任务",type= OperationTypeEnum.UPDATECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @ApiOperation("执行任务")
    @PostMapping("/runJobInfo")
    //    @RequiresPermissions("bdp:jobInfo:runJobInfo")
    public R runJobInfo(@ApiParam(value = "{'id':1,'last_run_time':'2019-01-01'}") @RequestBody JobInfo param) {
        checkIsMod(param);
        //LastRunTime里存储的是任务的参数
        if (param.getLastRunTime() == null) {
            param.setLastRunTime(DateTime.now());
        }
//        String time_hour = DateUtil.format(param.getLastRunTime(),"yyyy-MM-dd HH:mm:ss");

        JobInfo jobInfo = jobInfoService.getById(param.getId());
        if (!jobInfo.getEnable()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_DISABLE.getMessage());
        } else {

            //启用状态---让其变为运行中，不让其再执行
            jobInfo.setLastRunState(LastRunState.RUNNING.getCode());
            HdfsUtil hdfsUtil = null;
            if (jobInfoService.updateById(jobInfo)) {
                try {
//                    Map<String, String> params = new HashMap<>();
//                    params.put("time_hour", time_hour);

                    //初始化对应的节点记录信息，开始的。为了开启以后马上可以看到日志
                    jobNodeInfoService.initJobHistoryAndJobNodeHistory(param, "手动运行初始化数据");
                    //这个必须得有，因为开始节点不能停止
//                    ThreadUtil.sleep(4*1000);
                    //上传workflow并执行
                    logger.info("获取OozieClient");
                    OozieClient wc = new OozieClient(oozieConfig.getUrl());
                    hdfsUtil = new HdfsUtil(bdpJobConfig);
                    JobUtil jobUtil = new JobUtil(hdfsUtil, wc);
                    Long userId = ShiroUtils.getUserId();
                    Properties conf = jobUtil.createOrUpdateJob(userId, param.getId(), jobInfo, false);
                    String oozieJobId = jobUtil.runJob(conf);

                    jobInfo.setLastOozieJobId(oozieJobId);
                    jobInfoService.updateById(jobInfo);
                    JobRunHistory jobRunHistory = jobRunHistoryService.getBaseMapper().selectList(new QueryWrapper<JobRunHistory>().eq("job_info_id", jobInfo.getId()).orderByDesc("num")).get(0);
                    jobRunHistory.setOozieJobId(oozieJobId);
                    jobRunHistoryService.updateById(jobRunHistory);
                    JobNodeRunHistory jobNodeRunHistory = jobNodeRunHistoryService.getBaseMapper().selectList(new QueryWrapper<JobNodeRunHistory>().eq("job_info_id", jobInfo.getId()).orderByDesc("num")).get(0);
                    jobNodeRunHistory.setOozieJobId(oozieJobId);
                    jobNodeRunHistoryService.updateById(jobNodeRunHistory);

                    //jobInfoService.timeingRest(jobInfo, oozieJobId, oozieConfig, false);
                    return R.ok();
                } catch (Exception e) {
                    logger.error("error:{}", e.getMessage());
                    throw new RRException(DataDevelopmentBizExceptionEnum.SERVER_ERROR.getMessage());
                }finally {
                    if(null != hdfsUtil){
                        hdfsUtil.close();
                    }
                }
            }
            return R.ok();
        }
    }

    private void checkStopIsMod(JobInfo jobInfo) {
        if (jobInfo.getId() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_ID_NULL.getMessage());
        }
        if (!jobInfoService.checkExistById(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NOTEXISTED.getMessage());
        }
        if (!jobInfoService.checkPermission(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_PERMISSIOIN.getMessage());
        }
        if (jobInfoService.checkJobIsEable(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NO_EABLE_STOP.getMessage());
        }
    }

    private void checkIsMod(JobInfo jobInfo) {
        if (jobInfo.getId() == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_ID_NULL.getMessage());
        }
        if (!jobInfoService.checkExistById(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NOTEXISTED.getMessage());
        }
        if (!jobInfoService.checkPermission(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_PERMISSIOIN.getMessage());
        }
        if (jobInfoService.checkIsRun(jobInfo.getId())) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_RUN.getMessage());
        }
    }

    /**
     * 返回的是任务详情，对应job_info数据表
     */
    @ApiOperation("任务详情")
    @GetMapping("/info/{id}")
    @SysLog(value="任务详情",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
//    @RequiresPermissions("bdp:jobInfo:info")
    public R info(@PathVariable("id") Integer id) {
        JobInfo jobInfo = jobInfoService.getById(id);
        if (jobInfo != null) {
            fillInfo(jobInfo);
            List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>().eq("job_id", id));
            jobInfo.setJobNodeInfos(jobNodeInfos);
        }
        return R.okWithData(jobInfo);
    }


    /**
     * 保存任务节点信息
     * { "class": "GraphLinksModel",
     * "linkFromPortIdProperty": "fromPort",
     * "linkToPortIdProperty": "toPort",
     * "nodeDataArray": [
     * {"category":"Start", "text":"Start", "key":-1, "loc":"-93.5 353"},
     * {"text":"jjj", "key":-2, "loc":"45.5 400"},
     * {"category":"End", "text":"End", "key":-4, "loc":"335.5 364"},
     * {"text":"Step", "key":-6, "loc":"124.5 307"},
     * {"text":"Step", "key":-7, "loc":"243.5 325"},
     * {"text":"Step", "key":-8, "loc":"28.5 312"}
     * ],
     * "linkDataArray": [
     * {"from":-1, "to":-2, "fromPort":"R", "toPort":"L", "points":[-69.2266599078511,353,-59.2266599078511,353,-18.918089933173597,353,-18.918089933173597,400,21.390480041503906,400,31.390480041503906,400]},
     * {"from":-2, "to":-7, "fromPort":"R", "toPort":"L", "points":[59.609519958496094,400,69.6095199584961,400,139.61274337768555,400,139.61274337768555,330.3125747680664,209.615966796875,330.3125747680664,219.615966796875,330.3125747680664]},
     * {"from":-6, "to":-7, "fromPort":"R", "toPort":"L", "points":[148.384033203125,307,158.384033203125,307,184,307,184,319.6874252319336,209.615966796875,319.6874252319336,219.615966796875,319.6874252319336]},
     * {"from":-7, "to":-4, "fromPort":"R", "toPort":"L", "points":[267.384033203125,325,277.384033203125,325,291.3810020712919,325,291.3810020712919,364,305.3779709394588,364,315.3779709394588,364]},
     * {"from":-1, "to":-8, "fromPort":"R", "toPort":"L", "points":[-69.2266599078511,353,-59.2266599078511,353,-32.30534655548805,353,-32.30534655548805,312,-5.384033203125,312,4.615966796875,312]},
     * {"from":-8, "to":-6, "fromPort":"R", "toPort":"T", "points":[52.384033203125,312,62.384033203125,312,77.5,312,77.5,281.0622756958008,124.5,281.0622756958008,124.5,291.0622756958008]}
     * ]}
     */
    @ApiOperation("保存任务节点信息")
    @SysLog(value="保存任务节点信息",type= OperationTypeEnum.ADDCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/saveNode")
//    @RequiresPermissions("bdp:jobInfo:saveNode")
    public R saveNode(@ApiParam(value = "{'id':1,'param':{json}}") @RequestBody JobInfo param) {

        //现在改为，移动功能，则保存对应的功能
        checkIsMod(param);
        JobInfo job = jobInfoService.getById(param.getId());
        if (job == null) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NOTEXISTED.getMessage());
        } else {
            //为禁用的才可以修改
            if (job.getEnable()) {
                throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_ENABLE.getMessage());
            }
            job.setParam(param.getParam());
            jobInfoService.updateById(job);
        }
        return R.ok();
    }


//    @ApiOperation("更换为oozie版本而进行数据迁移")
//    @GetMapping("/saveNodeForOozie")
//    public R saveNodeForOozie(@ApiParam(value = "{'uid':5}") @RequestParam Integer uid) {
//        //定时任务列表：
//        List<Integer> jobID = new ArrayList<>();
//        List<JobInfo> jobInfos = jobInfoService.list(new QueryWrapper<JobInfo>()
//                .eq(null != uid, "create_per", uid));
//        Map<Integer, String> errorNode = new HashMap<>();
//        Map<Integer, String> errorJob = new HashMap<>();
//        Map<String, Object> errorLog = new HashMap<>();
//        Set<Integer> errorCauseByNode = new HashSet<>();
//        Map<Integer, String> jobStateDiff = new HashMap<>();
//        if (null == jobInfos) {
//            return R.ok();
//        }
//        for (JobInfo j : jobInfos) {
//            JobNodeInfo startNode = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("job_id", j.getId()));
//            if (null != startNode) {
//                List<JobNodeConf> jobNodeConfs = jobNodeConfService.list(new QueryWrapper<JobNodeConf>().eq("job_node_id", startNode.getId()).eq("type_id", 0));
//                JobNodeConfig jobNodeConfig = JobNodeConfig.listToJobNodeConfig(jobNodeConfs);
//                if (null != jobNodeConfig.getExpression() && "null".equals(jobNodeConfig.getExpression())) {
//                    jobID.add(j.getId());
//                    if (null == jobNodeConfig.getScheduleOrInterval()) {
//                        JobNodeConf jobNodeConf = new JobNodeConf();
//                        jobNodeConf.setJobNodeId(jobNodeConfig.getJobNodeId());
//                        jobNodeConf.setKey("scheduleOrInterval");
//                        jobNodeConf.setValue("1");
//                        jobNodeConfService.save(jobNodeConf);
//                    }
//                }
//            }
//
//            List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>().eq("job_id", j.getId()));
//            for (JobNodeInfo n : jobNodeInfos) {
//                JobNodeConfig jobNodeConfig = new JobNodeConfig();
//                if (n != null && null != n.getNodeKey()) {
//                    List<JobNodeConf> jobNodeConfs = jobNodeConfService.list(new QueryWrapper<JobNodeConf>().eq("job_node_id", n.getId()).eq("job_type", 0));
//                    jobNodeConfig = JobNodeConfig.listToJobNodeConfig(jobNodeConfs);
//
//                    // 将shell保存至hdfs
//                    Integer userId = j.getCreatePer();
//                    Integer jobId = n.getJobId();
//                    String nodeKey = n.getNodeKey();
//                    String nodeName = JobType.ObjOf(n.getTypeId()).getEnName();
//                    n.setJobNodeConfig(jobNodeConfig);
//                    String dst = "workflow-app/" + userId.toString() + "/" + jobId.toString() + "/" + nodeName + nodeKey.replace("-", "") + ".sh";
//                    try {
//                        System.out.println(jobId + ":" + n.toString());
//                        jobNodeInfoService.uploadShell(n, dst, oozieConfig);
//                    } catch (Exception e) {
//                        errorNode.put(n.getId(), e.getMessage());
//                        errorCauseByNode.add(j.getId());
//                        continue;
//                    }
//                }
//            }
//
//            try {
//                if (j.getEnable()) {
//                    if (j.getLastRunState().equals(LastRunState.RUNNING.getCode())) {
//                        stopJobInfo(j);
//                        disableJobInfo(j);
//                        if (!errorCauseByNode.contains(j.getId())) {
//                            enableJobInfo(j);
//                            runJobInfo(j);
//                        } else {
//                            jobStateDiff.put(j.getId(), "running-disable");
//                        }
//                    } else {
//                        disableJobInfo(j);
//                        if (!errorCauseByNode.contains(j.getId())) {
//                            enableJobInfo(j);
//                        } else {
//                            jobStateDiff.put(j.getId(), "enable-disable");
//                        }
//                    }
//                }
//            } catch (Exception e) {
//                errorJob.put(j.getId(), e.getMessage());
//                continue;
//            }
//        }
//
//        Map<String, Object> result = new HashMap<>();
//
//        if (null != errorNode) {
//            errorLog.put("errorNode", errorNode);
//            errorLog.put("jobStateDiff", jobStateDiff);
//        }
//        if (null != errorJob) {
//            errorLog.put("errorJob", errorJob);
//        }
//        if (null != errorLog) {
//            errorLog.put("定时任务", jobID);
//            result = errorLog;
//
//        } else {
//            result.put("定时任务", jobID);
//        }
//
//        return R.okWithData(result);
//
//    }


    @ApiOperation("workFlow版本初始化")
    @GetMapping("/workflowInit")
    public R workflowInit() {
        String url = " http://localhost:18080/data-development/bdp/jobInfo/generateWorkflow";

        JSONArray jsonArray = new JSONArray();
        List<SysUserEntity> sysUserEntities = sysUserService.list();
        Integer i =1;
        for (SysUserEntity sysUserEntity : sysUserEntities) {
//            if (!sysUserEntity.getUserId().equals(Long.valueOf(2))){
//                continue;
//            }
            logger.info("-----------------正在更新第"+(i++)+"个："+sysUserEntity.getUsername()+"["+sysUserEntity.getUserId()+"]的信息");
            SysUserTokenEntity sysUserTokenEntity = sysUserTokenService.getById(sysUserEntity.getUserId());
            JSONObject jsonObject = new JSONObject();
            try {
                HttpResponse resData = HttpRequest.get(url).header("token", sysUserTokenEntity.getToken()).execute();
                if (null == resData || resData.getStatus() != 200) {
                    jsonObject.put(sysUserEntity.getUserId().toString(), "更新该用户发生异常");
                    jsonArray.add(jsonObject);
                } else {
                    System.out.println(resData.body());
                    JSONObject jsonObject1 = JSONObject.parseObject(resData.body());
                    if (jsonObject1.get("data") != null) {
                        JSONObject jsonObject2 = (JSONObject) jsonObject1.get("data");
                        jsonObject.put(sysUserEntity.getUserId().toString(), jsonObject2);
                        jsonArray.add(jsonObject);
                    } else {
                        jsonObject.put(sysUserEntity.getUserId().toString(), jsonObject1);
                        jsonArray.add(jsonObject);
                    }
                }
            } catch (Exception e) {
//                //优化画布
//                String url2 = " http://localhost:18080/data-development/bdp/workMenu/rePoXAndY";
//                HttpRequest.get(url2).header("token", sysUserTokenEntity.getToken()).execute();
                throw new RRException("更新第"+(i++)+"个："+sysUserEntity.getUsername()+"["+sysUserEntity.getUserId()+"]的信息时出现异常出现异常！"+e.getMessage()+"---------------------"+jsonArray);
            }
//            //优化画布
//            String url2 = " http://localhost:18080/data-development/bdp/workMenu/rePoXAndY";
//            HttpRequest.get(url2).header("token", sysUserTokenEntity.getToken()).execute();

        }

        return R.okWithData(jsonArray);
    }


    @ApiOperation("workFlow版本初始化")
    @GetMapping("/generateWorkflow")
    public R generateWorkflow() {
        Map<Integer, String> errorNode = new HashMap<>();
        Map<Integer, String> errorJob = new HashMap<>();
        Map<String, Object> errorLog = new HashMap<>();
        Set<Integer> errorCauseByNode = new HashSet<>();
        Map<Integer, String> jobStateDiff = new HashMap<>();
        Map<Integer, String> jobStateDiffChange = new HashMap<>();

        //某用户旧的任务列表：
        List<Integer> jobID = new ArrayList<>();
        List<JobInfo> jobInfos = jobInfoService.list(new QueryWrapper<JobInfo>()
                .eq(null != ShiroUtils.getUserId(), "create_per", ShiroUtils.getUserId())
                .isNotNull("param"));

        if (null == jobInfos||jobInfos.size()==0) {
            return R.ok();
        }
        for (JobInfo j : jobInfos) {
//            logger.error("处理"+ShiroUtils.getUserId()+"的任务"+j.getProcId()+"/"+j.getId());
            ProcInfo procInfo = procInfoService.getById(j.getProcId());
            //判断是否正在运行中
            if (j.getLastRunState().equals(LastRunState.RUNNING.getCode())){
                try {
                    jobInfoService.stopJobInfo(j.getId(), j.getLastOozieJobId());
                } catch (Exception e) {
                    errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+":"+"停止运行失败"+e.getMessage());
                    continue;
                }
                j=jobInfoService.getById(j);
                j.setLastOozieJobId(null);
                jobInfoService.updateById(j);
                jobStateDiff.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+":"+"运行中---->停止状态");
            }

            //如果是定时任务，则需要停止定时任务
            JobNodeInfo startNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>()
                    .eq("job_id", j.getId())
                    .eq("type_id", 0));
            if (startNodeInfo!=null) {
                JobNodeConf jobNodeConf = jobNodeConfService.getOne(new QueryWrapper<JobNodeConf>().eq("`key`", "expression").eq("job_node_id", startNodeInfo.getId()).eq("job_type", 0));
                if (null != jobNodeConf && jobNodeConf.getValue() != null && !jobNodeConf.getValue().equals("null")) {
                    try {
                        jobInfoService.stopJobInfo(j.getId(), j.getLastCorOozieJobId());
                    } catch (Exception e) {
                        errorJob.put(j.getId(), procInfo.getName() + "/" + j.getProcId() + "/" + j.getName() + ":" + "停止定时调度失败" + e.getMessage());
                        continue;
                    }
                    j=jobInfoService.getById(j);
                    j.setLastCorOozieJobId(null);
                    j.setEnable(false);
                    jobInfoService.updateById(j);
                    if (jobStateDiff.get(j.getId()) != null) {
                        jobStateDiff.put(j.getId(), jobStateDiff.get(j.getId()) + ";---->停止调度状态");
                    } else {
                        jobStateDiff.put(j.getId(), procInfo.getName() + "/" + j.getProcId() + "/" + j.getName() + ":" + "定时调度启用中---->停止调度状态");
                    }
                }
            }



            //判断该项目是否有目录没有则创建
            List<WorkMenu> workMenus = workMenuService.list(new QueryWrapper<WorkMenu>()
                    .eq("proc_id", j.getProcId())
                    .eq("parent_id", 0));

            if (null == workMenus || workMenus.size() == 0) {
                //为项目添加一个根目录
                WorkMenuDto workMenuDto = new WorkMenuDto();
                try{
                    workMenuDto.setName("我的根目录1");
                    workMenuDto.setParentId(Long.valueOf(0));
                    workMenuDto.setProcId(j.getProcId());
                    workMenuService.addWorkFlowMenu(workMenuDto);
                }catch (Exception e){
                    errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+":"+"创建根目录失败"+e.getMessage());
                    continue;
                }


                WorkMenu workMenu1 = workMenuService.getOne(new QueryWrapper<WorkMenu>()
                        .eq("name", workMenuDto.getName())
                        .eq("parent_id", workMenuDto.getParentId())
                        .eq("proc_id", workMenuDto.getProcId()));
                //生成工作流及其节点以及节点连线
                generateWorkFlow(errorNode,errorJob,errorLog,errorCauseByNode,workMenu1,j,procInfo);
            } else {
                //获取一个根目录
                WorkMenu workMenu = workMenus.get(0);
                generateWorkFlow(errorNode,errorJob,errorLog,errorCauseByNode,workMenu,j,procInfo);
            }

            //改变状态
            String value = jobStateDiff.get(j.getId());
            if(StringUtils.isNotEmpty(value)){
                //查询以前是否是正常调度的方式
                if(null != startNodeInfo){
                    JobNodeConf jobNodeConf = jobNodeConfService.getOne(new QueryWrapper<JobNodeConf>().eq("`key`", "expression").eq("job_type", 0).eq("job_node_id", startNodeInfo.getId()));
                    if(null != jobNodeConf && StringUtils.isNotEmpty(jobNodeConf.getValue()) && !"null".equals(jobNodeConf.getValue())){
                        //改变对应的
                        try {
                            //对应的菜单保存对应的状态
                            WorkMenuJobRel workMenuJobRel = workMenuJobRelService.getOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_flow_id", j.getId()));
                            if(null != workMenuJobRel){
                                //改变状态
                                JobNodeConf jobNodeConf1 = jobNodeConfService.getOne(new QueryWrapper<JobNodeConf>().eq("`key`", "schedule_type").eq("job_type", 1).eq("job_node_id", workMenuJobRel.getWorkMenuId()));
                                if(null == jobNodeConf1){
                                    jobNodeConf1 = new JobNodeConf();
                                }
                                jobNodeConf1.setKey("schedule_type");
                                jobNodeConf1.setJobType(1);
                                jobNodeConf1.setValue("1");
                                jobNodeConf1.setJobNodeId(workMenuJobRel.getWorkMenuId().intValue());

                                JobNodeConf jobNodeConf2 = new JobNodeConf();
                                jobNodeConf2.setKey("scheduleOrInterval");
                                jobNodeConf2.setJobType(1);
                                jobNodeConf2.setValue("1");
                                jobNodeConf2.setJobNodeId(workMenuJobRel.getWorkMenuId().intValue());

                                JobNodeConf jobNodeConf3 = new JobNodeConf();
                                jobNodeConf3.setKey("expression");
                                jobNodeConf3.setJobType(1);
                                jobNodeConf3.setValue(jobNodeConf.getValue());
                                jobNodeConf3.setJobNodeId(workMenuJobRel.getWorkMenuId().intValue());

                                List<JobNodeConf> olderList = new ArrayList<>();
                                olderList.add(jobNodeConf1);
                                olderList.add(jobNodeConf2);
                                olderList.add(jobNodeConf3);
                                jobNodeConfService.upsertKVByJobNodeId(olderList);
                            }


                            //启动workflow
                            workMenuService.updateRouteWorkFlowXml(j.getId());

                        }catch (Exception e){
                            logger.error("生成对应的定时调度异常！jobId:{},error:{}",j.getId(),e.getMessage());
                            jobStateDiffChange.put(j.getId(),"改为定时调度异常！");
                        }
                    }
                }
            }
        }


        Map<String, Object> result = new HashMap<>();

        if (null != errorNode) {
            errorLog.put("errorNode", errorNode);
            errorLog.put("jobStateDiff", jobStateDiff);
            errorLog.put("jobStateDiffChange", jobStateDiffChange);
        }
        if (null != errorJob) {
            errorLog.put("errorJob", errorJob);
        }
        if (null != errorLog) {
            errorLog.put("定时任务", jobID);
            result = errorLog;

        } else {
            result.put("定时任务", jobID);
        }

        return R.okWithData(result);
    }


    public void generateWorkFlow(Map<Integer, String> errorNode, Map<Integer, String> errorJob,Map<String, Object> errorLog,Set<Integer> errorCauseByNode,WorkMenu workMenu1,JobInfo j,ProcInfo procInfo){
        WorkMenu workMenu = new WorkMenu();
        //添加工作流目录以及工作流与目录间的依赖
        try{
            workMenu = workMenuService.addOldWorkFlowMenu(workMenu1, j);
            if (workMenu==null){
                errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+":"+"创建工作流目录失败,为空");
                return;
            }
        }catch (Exception e){
            errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+":"+"创建工作流目录发生异常"+e.getMessage());
            return;
        }

        //验证节点的名字是否重复
        WorkMenuJobRel workMenuJobRel = workMenuJobRelService.getOne(new QueryWrapper<WorkMenuJobRel>().eq("work_type", 1).eq("work_menu_id", workMenu.getId()));
        if (null == workMenuJobRel) {
            errorNode.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+":"+"创建工作流节点失败，由于工作流目录依赖不存在,该工作流下的所有节点未进行同步");
            return;
        }

        //添加工作流的任务节点
        List<JobNodeInfo> jobNodeInfos = jobNodeInfoService.list(new QueryWrapper<JobNodeInfo>()
                .eq("job_id", j.getId())
                .ne("type_id", 0).ne("type_id", 3).ne("type_id", 5));
        if (null == jobNodeInfos||jobNodeInfos.size()==0) {
            return;
        }

        for (JobNodeInfo jobNodeInfo : jobNodeInfos) {
            logger.debug("处理" + ShiroUtils.getUserId() + "的任务" + j.getProcId() + "/" + j.getId() + "/" + jobNodeInfo.getId());
            //保存节点配置信息
            //数据同步节点需要初始化的信息
            switch (jobNodeInfo.getTypeId()) {
                case 1:
                    List<String> inputKeys = Arrays.asList("hightMaxConcurrent", "jobType");
                    List<String> inputValues = Arrays.asList("2", "0");
                    for (int i = 0; i < inputKeys.size(); i++) {
                        if (jobNodeConfService.selectKVByJobNodeIdAndKey(jobNodeInfo.getId(), inputKeys.get(i), 0) == null) {
                            try {
                                JobNodeConf jobNodeConf = new JobNodeConf();
                                jobNodeConf.setKey(inputKeys.get(i));
                                jobNodeConf.setValue(inputValues.get(i));
                                jobNodeConf.setJobNodeId(jobNodeInfo.getId());
                                jobNodeConf.setJobType(0);
                                jobNodeConfService.save(jobNodeConf);
                            }catch (Exception e){
                                errorNode.put(jobNodeInfo.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+"/"+jobNodeInfo.getName()+":"+"添加"+inputKeys.get(i)+"的配置信息出错"+e.getMessage());
                            }

                        }
                    }
                    //固定配置
                    List<String> outputKeys = Arrays.asList("resource_dm", "resource_dc", "resource_em", "resource_ec");
                    List<String> outputValues = Arrays.asList("1", "1", "4", "2");
                    List<JobNodeConf> list = new ArrayList<>();
                    for (int i = 0; i < outputKeys.size(); i++) {
                        try {
                            JobNodeConf jobNodeConf1 = jobNodeConfService.selectKVByJobNodeIdAndKey(jobNodeInfo.getId(), outputKeys.get(i), 0);
                            if (jobNodeConf1 == null) {
                                JobNodeConf jobNodeConf = new JobNodeConf();
                                jobNodeConf.setKey(outputKeys.get(i));
                                jobNodeConf.setValue(outputValues.get(i));
                                jobNodeConf.setJobNodeId(jobNodeInfo.getId());
                                jobNodeConf.setJobType(0);
                                jobNodeConfService.save(jobNodeConf);
                            }else{
                                if(StringUtils.isEmpty(jobNodeConf1.getValue()) || "null".equals(jobNodeConf1.getValue())){
                                    jobNodeConf1.setValue(outputValues.get(i));
                                    list.add(jobNodeConf1);
                                }
                            }
                        }catch (Exception e){
                            errorNode.put(jobNodeInfo.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+"/"+jobNodeInfo.getName()+":"+"添加"+outputKeys.get(i)+"的配置信息出错"+e.getMessage());
                        }
                    }
                    if(CollectionUtil.isNotEmpty(list)){
                        jobNodeConfService.upsertKVByJobNodeId(list);
                    }
                    break;
                case 2:
                    //sql计算节点需要初始化的信息
                    try{
                        if (jobNodeConfService.selectKVByJobNodeIdAndKey(jobNodeInfo.getId(), "jobType", 0) == null) {

                            JobNodeConf jobNodeConf = new JobNodeConf();
                            jobNodeConf.setKey("jobType");
                            jobNodeConf.setValue("0");
                            jobNodeConf.setJobNodeId(jobNodeInfo.getId());
                            jobNodeConf.setJobType(0);
                            jobNodeConfService.save(jobNodeConf);
                        }
                    }catch (Exception e){
                        errorNode.put(jobNodeInfo.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+"/"+jobNodeInfo.getName()+":"+"添加jobType的配置信息出错"+e.getMessage());
                    }
                    break;
                case 4:
                    //数据推送节点需要初始化的信息
                    List<String> outputKeys1 = Arrays.asList("resource_dm", "resource_dc", "resource_em", "resource_ec", "jobType");
                    List<String> outputValues1 = Arrays.asList("1", "1", "4", "2", "0");
                    List<JobNodeConf> list1 = new ArrayList<>();
                    for (int i = 0; i < outputKeys1.size(); i++) {
                        try {
                            JobNodeConf jobNodeConf1 = jobNodeConfService.selectKVByJobNodeIdAndKey(jobNodeInfo.getId(), outputKeys1.get(i), 0);
                            if ( jobNodeConf1 == null) {
                                JobNodeConf jobNodeConf =   new JobNodeConf();
                                jobNodeConf.setKey(outputKeys1.get(i));
                                jobNodeConf.setValue(outputValues1.get(i));
                                jobNodeConf.setJobNodeId(jobNodeInfo.getId());
                                jobNodeConf.setJobType(0);
                                jobNodeConfService.save(jobNodeConf);
                            }else {
                                if(StringUtils.isEmpty(jobNodeConf1.getValue()) || "null".equals(jobNodeConf1.getValue())){
                                    jobNodeConf1.setValue(outputValues1.get(i));
                                    list1.add(jobNodeConf1);
                                }
                            }
                        }catch (Exception e){
                            errorNode.put(jobNodeInfo.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+"/"+jobNodeInfo.getName()+":"+"添加"+outputKeys1.get(i)+"的配置信息出错"+e.getMessage());
                        }
                    }
                    if(CollectionUtil.isNotEmpty(list1)){
                        jobNodeConfService.upsertKVByJobNodeId(list1);
                    }
                    break;
            }

            //保存节点信息
            jobNodeInfo.setEnable(JobStatus.ACTIVE.getCode());

//                    jobNodeInfo.setNodeKey(null);
//                    jobNodeInfo.setPNodeKey(null);
            jobNodeInfoService.updateById(jobNodeInfo);
            List<JobNodeConf> confs = jobNodeConfService.list(new QueryWrapper<JobNodeConf>()
                    .eq("job_type",0)
                    .eq("job_node_id",jobNodeInfo.getId()));
            JobNodeConfig jobNodeConfig = JobNodeConfig.listToJobNodeConfig(confs);
            jobNodeInfo.setJobNodeConfig(jobNodeConfig);

            WorkMenuJobRel workMenuJobRel2 = workMenuJobRelService.getOne(new QueryWrapper<WorkMenuJobRel>()
                    .eq("work_flow_id",jobNodeInfo.getId())
                    .eq("work_type",0)
                    .eq("work_menu_id",workMenuJobRel.getWorkMenuId()));
            if (workMenuJobRel2==null){
                //添加节点关联表数据
                WorkMenuJobRel workMenuJobRel1 = new WorkMenuJobRel();
                BeanUtils.copyProperties(workMenuJobRel, workMenuJobRel1);
                workMenuJobRel1.setId(null);
                workMenuJobRel1.setWorkFlowId(Long.valueOf(jobNodeInfo.getId()));
                workMenuJobRel1.setWorkType(0);
                workMenuJobRelService.save(workMenuJobRel1);
            }


            //保存节点的shell到hdfs
            // 将shell保存至hdfs
            Long userId = ShiroUtils.getUserId();
            Integer jobId = jobNodeInfo.getJobId();
            Integer nodeId = jobNodeInfo.getId();
            String nodeName = JobType.ObjOf(jobNodeInfo.getTypeId()).getEnName();
            String dst = "workflow-app/" + WorkFlowType.WORKFLOW.getCode() + "/" + j.getProcId() + "/" + jobId.toString() + "/" + nodeName + "_" + nodeId + ".sh";
            if (jobNodeInfo.getId().equals(9)){
                try {
                    jobNodeInfoService.uploadShell(jobNodeInfo, dst, oozieConfig);
                }catch (Exception e){
                    errorNode.put(jobNodeInfo.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+"/"+jobNodeInfo.getName()+":"+"上传任务节点shell没有成功"+e.getMessage());
                }
            }else{
                try {
                    jobNodeInfoService.uploadShell(jobNodeInfo, dst, oozieConfig);
                }catch (Exception e){
                    errorNode.put(jobNodeInfo.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+"/"+jobNodeInfo.getName()+":"+"上传任务节点shell没有成功"+e.getMessage());
                }
            }



        }



        //保存节点的连线
        JSONObject jsonObject = JSONObject.parseObject(j.getParam());
        JSONArray nodeDataArray = jsonObject.getJSONArray("nodeDataArray");
        JSONArray linkDataArray = jsonObject.getJSONArray("linkDataArray");

        List<JSONObject> linkJsonObjects = linkDataArray.toJavaList(JSONObject.class);
        List<JSONObject> nodeJsonObjects = nodeDataArray.toJavaList(JSONObject.class);
        List<JobLinkVO> edges = new ArrayList<>();
        List<JobNodeInfoVO> nodes = new ArrayList<>();
        try {
            linkJsonObjects.forEach(d -> {
                String fromKey = d.getString("from");
                String toKey = d.getString("to");
                JobNodeInfo fromNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("node_key", fromKey).eq("job_id", j.getId())
                        .ne("type_id", 0).ne("type_id", 3).ne("type_id", 5));
                JobNodeInfo toNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("node_key", toKey).eq("job_id", j.getId())
                        .ne("type_id", 0).ne("type_id", 3).ne("type_id", 5));

                if (fromNodeInfo != null && toNodeInfo != null) {
                    JobLinkVO edge = new JobLinkVO();
                    edge.setSrcNodeId(fromNodeInfo.getId());
                    edge.setDstNodeId(toNodeInfo.getId());
                    edges.add(edge);
                }
            });
        }catch (Exception e){
            errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+":"+"解析edges出错"+e.getMessage());
            return;
        }

        try {
            nodeJsonObjects.forEach(n -> {
                String nodeKey = n.getString("key");
                String posX = n.getString("loc").split(" ")[0];
                String posY = n.getString("loc").split(" ")[1];

                JobNodeInfo jobNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("node_key", nodeKey).eq("job_id", j.getId())
                        .ne("type_id", 0).ne("type_id", 3).ne("type_id", 5));
                if (jobNodeInfo != null) {
                    JobNodeInfoVO node = new JobNodeInfoVO();
                    node.setId(jobNodeInfo.getId());
                    node.setName(jobNodeInfo.getName());
                    node.setPosX(Double.parseDouble(posX.replace("-", "")));
                    node.setPosY(Double.parseDouble(posY.replace("-", "")));
                    node.setTypeId(jobNodeInfo.getTypeId());
                    nodes.add(node);
                }

            });
        }catch (Exception e){
            errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+":"+"解析nodes出错"+e.getMessage());
            return;
        }



        //获取原来开始节点的配置
        JobNodeInfo startNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("type_id", 0).eq("job_id", j.getId()));
        JobNodeConfig jobNodeConfig = new JobNodeConfig();
        if (startNodeInfo != null) {
            JobNodeConf expressionConf = jobNodeConfService.selectKVByJobNodeIdAndKey(startNodeInfo.getId(), "expression", 0);
            if (expressionConf != null) {
                jobNodeConfig.setExpression(expressionConf.getValue());
            }
            JobNodeConf scheduleOrIntervalConf = jobNodeConfService.selectKVByJobNodeIdAndKey(startNodeInfo.getId(), "scheduleOrInterval", 0);
            if (scheduleOrIntervalConf != null) {
                jobNodeConfig.setScheduleOrInterval(Integer.parseInt(scheduleOrIntervalConf.getValue()));
            }
            JobNodeConf scheduleTypeConf = jobNodeConfService.selectKVByJobNodeIdAndKey(startNodeInfo.getId(), "schedule_type", 0);
            if (scheduleTypeConf != null) {
                jobNodeConfig.setSchedule_type(0);//暂停调度
            }
            JobNodeConf sysParamConf = jobNodeConfService.selectKVByJobNodeIdAndKey(startNodeInfo.getId(), "sysParam", 0);
            if (sysParamConf != null) {
                if (sysParamConf.getValue().equals("[]")) {
                    jobNodeConfig.setSysParam(new ArrayList<>());
                } else {
                    JSONArray jsonArray = JSONArray.parseArray(sysParamConf.getValue());
                    List<Map<String, Object>> list = new ArrayList<>();
                    jsonArray.forEach(json -> {
                        JSONObject jsonObject1 = (JSONObject) json;
                        Map<String, Object> map = jsonObject1.getInnerMap();
                        list.add(map);
                    });
                    jobNodeConfig.setSysParam(list);
                }
            }
        }

        try {
            JobLinkAndJobNodeDto jobLinkAndJobNodeDto = new JobLinkAndJobNodeDto();
            jobLinkAndJobNodeDto.setEdges(PojoUtils.listConvert(JobLinkDto.class, edges));
            jobLinkAndJobNodeDto.setNodes(PojoUtils.listConvert(JobNodeInfoDto.class, nodes));
            jobLinkAndJobNodeDto.setJobNodeConfig(jobNodeConfig);
            jobLinkAndJobNodeDto.setProcId(j.getProcId());
            jobLinkAndJobNodeDto.setWorkFlowMenuId(workMenu.getId());
            workMenuService.updateWorkJob(jobLinkAndJobNodeDto);
        }catch (Exception e){
            errorJob.put(j.getId(),procInfo.getName()+"/"+j.getProcId()+"/"+j.getName()+"/"+j.getId()+":"+"任务连线出错"+e.getMessage());
            return;
        }
    }
}
